package com.me.kafka.service.impl;

import com.me.kafka.exception.KafkaProduceMessageException;
import com.me.kafka.service.ProduceKafkaMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import javax.annotation.Resource;

/**
 * 生产Kafka消息
 *
 * @author Wuzhiming
 */
@Service
public class ProducerKafkaMessageImpl implements ProduceKafkaMessage {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    @Transactional
    public ListenableFuture<SendResult<String, String>> sendMessage(String topic, String message) {
        return kafkaTemplate.send(topic, message);
    }

    @Override
    @Transactional
    public void sendMessageRollback(String topic, String message, SuccessCallback successCallback, FailureCallback failureCallback) {
        kafkaTemplate.send(topic, message).addCallback(successCallback, failureCallback);
    }

    @Override
    @Transactional
    public void sendMessageRollback(String topic, String message, SuccessCallback successCallback) {
        kafkaTemplate.send(topic, message).addCallback(successCallback, null);
    }

}
